﻿using Confluent.Kafka;
using MySqlSync.Core.Interfaces;
using MySqlSync.Core.Models;
using System.Text.Json;

namespace MySqlSync.Kafka
{
    /// <summary>
    /// Kafka生产者服务
    /// </summary>
    public class KafkaProducerService : IKafkaProducerService
    {
        private readonly IProducer<string, string> _producer;
        private readonly string _topic;
        private readonly JsonSerializerOptions _jsonOptions = new JsonSerializerOptions
        {
            PropertyNameCaseInsensitive = true,
            WriteIndented = false
        };

        /// <summary>
        /// 构造函数
        /// </summary>
        public KafkaProducerService(KafkaConfig config)
        {
            if (config == null)
                throw new ArgumentNullException(nameof(config));

            if (string.IsNullOrEmpty(config.BootstrapServers))
                throw new ArgumentException("Kafka bootstrap servers 不能为空", nameof(config));

            if (string.IsNullOrEmpty(config.Topic))
                throw new ArgumentException("Kafka topic 不能为空", nameof(config));

            var producerConfig = new ProducerConfig
            {
                BootstrapServers = config.BootstrapServers,
                Acks = Acks.All, // 确保消息被所有ISR副本确认
                //RetryCount = 3,
                RetryBackoffMs = 1000
            };

            _producer = new ProducerBuilder<string, string>(producerConfig).Build();
            _topic = config.Topic;
        }

        /// <summary>
        /// 发送同步记录到Kafka
        /// </summary>
        public async Task SendMessageAsync(SyncRecord record)
        {
            if (record == null)
                throw new ArgumentNullException(nameof(record));

            if (string.IsNullOrEmpty(record.TableName))
                throw new ArgumentException("同步记录的表名不能为空", nameof(record));

            if (string.IsNullOrEmpty(record.RecordId))
                throw new ArgumentException("同步记录的ID不能为空", nameof(record));

            try
            {
                var key = $"{record.TableName}:{record.RecordId}";
                var value = JsonSerializer.Serialize(record, _jsonOptions);

                var deliveryResult = await _producer.ProduceAsync(
                    _topic,
                    new Message<string, string> { Key = key, Value = value }
                );

                // 可以添加日志记录成功发送的消息
            }
            catch (ProduceException<string, string> ex)
            {
                Console.WriteLine($"发送消息到Kafka失败: {ex.Error.Reason}");
                throw;
            }
        }

        /// <summary>
        /// 确保所有消息都已发送
        /// </summary>
        public Task FlushAsync()
        {
            _producer.Flush();
            return Task.CompletedTask;
        }

        /// <summary>
        /// 释放资源
        /// </summary>
        public void Dispose()
        {
            _producer?.Flush(TimeSpan.FromSeconds(10));
            _producer?.Dispose();
        }
    }

}
